package com.supreme.commonrocketmq.controller;

import com.alibaba.fastjson.JSONObject;
import com.supreme.common.entityRocketmq.MQExample;
import com.supreme.common.entityRocketmq.MQLogger;
import com.supreme.common.entityRocketmq.MQOrder;
import com.supreme.commonrocketmq.config.props.RocketmqProps;
import com.supreme.commonboot.utils.AjaxResp;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;

@RestController
@RequestMapping(value = "rocketmq")
public class RocketmqController {

    @Resource
    private RocketmqProps rocketmqProps;

    @Resource
    private DefaultMQProducer exampleProducer;

    @Resource
    private DefaultMQProducer orderProducer;

    @Resource
    private DefaultMQProducer loggerProducer;

    @PostMapping(value = "sendExampleMessage")
    public AjaxResp sendExampleMessage(@RequestParam(value = "tags", required = false) String tags,
                                       @RequestParam(value = "sendType", defaultValue = "1", required = false) Integer sendType,
                                       @RequestParam(value = "condition", defaultValue = "1", required = false) String condition,
                                       @RequestBody MQExample payload) {
        try {
            Message message = new Message();
            message.setTopic(rocketmqProps.getExample().getName());
            if (StringUtils.isNotBlank(tags)) {
                message.setTags(tags);
            }

            if (StringUtils.isNotBlank(condition)) {
                message.putUserProperty("condition", condition);
            }

            byte[] jsonBytes = JSONObject.toJSONBytes(payload);
            message.setBody(jsonBytes);

            if (sendType == 1) {
                SendResult sendResult = exampleProducer.send(message);
                System.out.println("sendResult: " + sendResult.getSendStatus());
            } else if (sendType == 2) {
                exampleProducer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("onSuccess: " + sendResult.getSendStatus());
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        System.out.println("onException: " + throwable.getMessage());
                    }
                });
            } else if (sendType == 3) {
                exampleProducer.sendOneway(message);
            }
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }

        return AjaxResp.instance();
    }

    @PostMapping(value = "sendExampleListMessage")
    public AjaxResp sendExampleListMessage(@RequestParam(value = "tags", required = false) String tags,
                                           @RequestParam(value = "sendType", defaultValue = "1", required = false) Integer sendType,
                                           @RequestParam(value = "condition", defaultValue = "1", required = false) String condition,
                                           @RequestBody List<MQExample> payload) {
        try {
            List<Message> messageList = payload.stream().map(new Function<MQExample, Message>() {
                @Override
                public Message apply(MQExample mqExample) {
                    Message message = new Message();
                    message.setTopic(rocketmqProps.getExample().getName());
                    if (StringUtils.isNotBlank(tags)) {
                        message.setTags(tags);
                    }

                    if (StringUtils.isNotBlank(condition)) {
                        message.putUserProperty("condition", condition);
                    }

                    byte[] jsonBytes = JSONObject.toJSONBytes(mqExample);
                    message.setBody(jsonBytes);
                    return message;
                }
            }).collect(Collectors.toList());

            SendResult sendResult = exampleProducer.send(messageList);
            System.out.println("sendResult: " + sendResult);
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }

        return AjaxResp.instance();
    }

    @PostMapping(value = "sendOrderMessage")
    public AjaxResp sendExampleMessage(@RequestBody MQOrder payload) {
        try {
            Message message = new Message();
            message.setTopic(rocketmqProps.getOrder().getName());

            byte[] jsonBytes = JSONObject.toJSONBytes(payload);
            message.setBody(jsonBytes);
            SendResult sendResult = orderProducer.send(message);
            System.out.println("sendResult: " + sendResult);
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
//            e.printStackTrace();
            return AjaxResp.instance(-1, e.getMessage());
        }

        return AjaxResp.instance();
    }

    @PostMapping(value = "sendLoggerMessage")
    public AjaxResp sendLoggerMessage(@RequestBody MQLogger payload) {
        try {
            Message message = new Message();
            message.setTopic(rocketmqProps.getLogger().getName());

            byte[] jsonBytes = JSONObject.toJSONBytes(payload);
            message.setBody(jsonBytes);
            SendResult sendResult = loggerProducer.send(message);
            System.out.println("sendResult: " + sendResult);
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }

        return AjaxResp.instance();
    }
}
